use bytes::Bytes;
use crossbeam_skiplist::SkipMap;
use memmap2::MmapOptions;
use mywal::Wal;
use parking_lot::{RwLock, RwLockReadGuard};
use serde::{Deserialize, Serialize};
use sled::{Db, IVec};
use std::cell::RefCell;
use std::collections::BTreeSet;
use std::fs::File;
use std::io::{Read, Seek};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::{fs, io, mem};

const FILE_NAME: &str = "minibitcask.data";
const MERGE_FILE_NAME: &str = "minibitcask.data.merge";
const ENTRY_HEADER_SIZE: usize = 10;

#[derive(Debug)]
struct Entry {
    key: Vec<u8>,
    value: Vec<u8>,
    key_size: u32,
    value_size: u32,
    mark: u16,
}

impl Entry {
    fn new(key: Vec<u8>, value: Vec<u8>, mark: u16) -> Self {
        let key_size = key.len() as u32;
        let value_size = value.len() as u32;
        Entry {
            key,
            value,
            key_size,
            value_size,
            mark,
        }
    }

    fn size(&self) -> usize {
        ENTRY_HEADER_SIZE + self.key_size as usize + self.value_size as usize
    }

    fn encode(&self) -> Vec<u8> {
        let mut buf = vec![0; self.size()];
        buf[0..4].copy_from_slice(&self.key_size.to_be_bytes());
        buf[4..8].copy_from_slice(&self.value_size.to_be_bytes());
        buf[8..10].copy_from_slice(&self.mark.to_be_bytes());
        buf[ENTRY_HEADER_SIZE..ENTRY_HEADER_SIZE + self.key.len()].copy_from_slice(&self.key);
        buf[ENTRY_HEADER_SIZE + self.key.len()..].copy_from_slice(&self.value);
        buf
    }

    fn decode(buf: &[u8]) -> Result<Self, io::Error> {
        if buf.len() < ENTRY_HEADER_SIZE {
            return Err(io::Error::new(io::ErrorKind::NotFound, "Wrong buffer size"));
        }

        let key_size = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]);
        let value_size = u32::from_be_bytes([buf[4], buf[5], buf[6], buf[7]]);
        let mark = u16::from_be_bytes([buf[8], buf[9]]);

        Ok(Entry {
            key: vec![],
            value: vec![],
            key_size,
            value_size,
            mark,
        })
    }
}

struct DBFile {
    file: File,
    offset: u64,
    file_path: String,
}

impl DBFile {
    fn new_internal(file_name: &str) -> Result<Self, std::io::Error> {
        let file = fs::OpenOptions::new()
            .create(true)
            .read(true)
            .write(true)
            .open(file_name)?;

        let metadata = file.metadata()?;
        let offset = metadata.len();

        Ok(DBFile {
            file,
            offset,
            file_path: file_name.to_string(),
        })
    }

    fn new(path: &str) -> Result<Self, std::io::Error> {
        let file_name = PathBuf::from(path).join(FILE_NAME);
        DBFile::new_internal(file_name.to_str().unwrap())
    }

    fn new_merge_db_file(path: &str) -> Result<Self, std::io::Error> {
        let file_name = PathBuf::from(path).join(MERGE_FILE_NAME);
        DBFile::new_internal(file_name.to_str().unwrap())
    }

    fn read(&mut self, offset: u64) -> Result<Entry, std::io::Error> {
        let mut buf = vec![0; ENTRY_HEADER_SIZE];
        self.file.read_exact_at_offset(&mut buf, offset)?;
        let mut entry = Entry::decode(&buf)?;

        let mut key = vec![0; entry.key_size as usize];
        let mut value = vec![0; entry.value_size as usize];

        if entry.key_size > 0 {
            self.file
                .read_exact_at_offset(&mut key, offset + ENTRY_HEADER_SIZE as u64)?;
        }

        if entry.value_size > 0 {
            self.file.read_exact_at_offset(
                &mut value,
                offset + ENTRY_HEADER_SIZE as u64 + entry.key_size as u64,
            )?;
        }

        entry.key = key;
        entry.value = value;

        Ok(entry)
    }

    fn write(&mut self, entry: &Entry) -> Result<(), std::io::Error> {
        let encoded = entry.encode();
        self.file.write_all_at_offset(&encoded, self.offset)?;
        self.offset += entry.size() as u64;
        Ok(())
    }
}

trait FileReadAt {
    fn read_exact_at_offset(&mut self, buf: &mut [u8], offset: u64) -> Result<(), std::io::Error>;
}

trait FileWriteAt {
    fn write_all_at_offset(&mut self, buf: &[u8], offset: u64) -> Result<(), std::io::Error>;
}

impl FileReadAt for File {
    fn read_exact_at_offset(&mut self, buf: &mut [u8], offset: u64) -> Result<(), std::io::Error> {
        self.seek(std::io::SeekFrom::Start(offset))?;
        self.read_exact(buf)?;
        Ok(())
    }
}

impl FileWriteAt for File {
    fn write_all_at_offset(&mut self, buf: &[u8], offset: u64) -> Result<(), std::io::Error> {
        self.seek(std::io::SeekFrom::Start(offset))?;
        self.set_len(offset + buf.len() as u64)?;
        let mut mmap = unsafe { MmapOptions::new().offset(offset).map_mut(&*self)? };
        mmap[..].copy_from_slice(buf);
        mmap.flush()?;
        Ok(())
    }
}

#[derive(Debug, Serialize, Deserialize)]
enum LogMark {
    Write,
    Delete,
}

#[derive(Debug, Serialize, Deserialize)]
struct LogValue {
    log_mark: LogMark,
    value: Option<Vec<u8>>,
}

impl LogValue {
    fn encode(&self) -> Vec<u8> {
        bincode::serialize(self).unwrap()
    }
    fn decode(bytes: &[u8]) -> Self {
        bincode::deserialize(bytes).unwrap()
    }
}

struct MemTable {
    id: u64,
    inner: SkipMap<Bytes, Bytes>,
    wal: Wal,
    size: RefCell<usize>,
    mem_table_size: u32,
}

impl MemTable {
    fn new<P: AsRef<Path>>(
        path: P,
        id: u64,
        mem_table_size: u32,
    ) -> io::Result<Self> {
        Ok(Self {
            id: id,
            inner: SkipMap::new(),
            wal: Wal::create(path)?,
            size: RefCell::new(0),
            mem_table_size,
        })
    }

    fn get(&self, key: &[u8]) -> Option<Bytes> {
        self.inner.get(key).map(|v| v.value().clone())
    }

    fn put(&self, key: &[u8], value: &[u8]) -> io::Result<()> {
        self.wal.write(key, value)?;
        let mut size = self.size.borrow_mut();
        if self.inner.contains_key(key) {
            self.inner
                .insert(Bytes::copy_from_slice(key), Bytes::copy_from_slice(value));
            let oldvsize = self.inner.get(key).unwrap().value().len();
            *size -= oldvsize;
            *size += value.len();
        } else {
            self.inner
                .insert(Bytes::copy_from_slice(key), Bytes::copy_from_slice(value));
            *size += key.len() + value.len()
        }
        Ok(())
    }
    fn size(&self) -> usize {
        *self.size.borrow()
    }

    fn is_full(&self) -> bool {
        self.size() >= self.mem_table_size as usize
    }
}

pub struct MiniBitcask {
    active_memtable: Arc<RwLock<MemTable>>,
    immu_memtables: Arc<RwLock<Vec<Arc<RwLock<MemTable>>>>>,
    index: Db,
    vlog: Arc<RwLock<DBFile>>,
    dir_path: String,
    memtable_nums: i32,
    mem_table_size: u32,
}

fn u64_to_bytes(value: u64) -> [u8; 8] {
    value.to_le_bytes()
}
fn bytes_to_u64(bytes: &[u8]) -> u64 {
    let array: [u8; 8] = bytes.try_into().unwrap();
    u64::from_le_bytes(array)
}

impl MiniBitcask {
    pub fn open(dir_path: &Path) -> Result<Self, io::Error> {
        OpenOptions::new(dir_path).open()
    }

    pub fn merge(&mut self) -> Result<(), io::Error> {
        let db_file_lock = Arc::clone(&self.vlog);
        let mut db_file = db_file_lock.write();

        if db_file.offset == 0 {
            return Ok(());
        }

        let mut valid_entries = Vec::new();
        let mut offset = 0;

        loop {
            match db_file.read(offset) {
                Ok(e) => {
                    let key = String::from_utf8_lossy(&e.key).to_string();
                    let size = e.size();
                    if let Some(off) = self.index.get(&key)? {
                        if bytes_to_u64(&off) == offset {
                            valid_entries.push(e);
                        }
                    }
                    offset += size as u64;
                }
                Err(err) => {
                    if err.kind() == io::ErrorKind::UnexpectedEof {
                        break;
                    } else {
                        return Err(err);
                    }
                }
            }
        }

        let mut merge_db_file = DBFile::new_merge_db_file(&self.dir_path)?;
        let merge_db_file_name = merge_db_file.file_path.clone();
        let idx_path = PathBuf::from(&self.dir_path).join("idx");
        let new_idx_path = PathBuf::from(&self.dir_path).join("idx.merge");
        let merged_indexes = sled::open(&new_idx_path)?;

        for entry in valid_entries {
            let write_off = merge_db_file.offset;
            merge_db_file.write(&entry)?;

            merged_indexes.insert(
                String::from_utf8_lossy(&entry.key).into_owned(),
                &u64_to_bytes(write_off),
            )?;
        }

        let db_file_name = db_file.file_path.clone();

        fs::remove_file(&db_file_name)?;
        fs::rename(
            &merge_db_file_name,
            PathBuf::from(&self.dir_path).join(FILE_NAME),
        )?;
        fs::remove_file(&idx_path)?;
        fs::rename(&new_idx_path, &idx_path)?;

        let new_db_file = DBFile::new(&self.dir_path)?;
        *db_file = new_db_file;
        self.index = sled::open(idx_path)?;
        Ok(())
    }

    fn write(&self, key: &[u8], log: LogValue) -> io::Result<()> {
        /*
            1. 判断键值对占用大小
            2. 如果键值对写入活跃内存表后活跃内存表满了则将活跃内存表归档进只读内存表队列中
            3. 如果只读内存表队列满了则将队头归档进index和vlog
        */
        let active_table = self.active_memtable.write();
        active_table.put(key, &log.encode())?;
        if active_table.is_full() {
            drop(active_table);
            let immu_tables = self.immu_memtables.read();
            if immu_tables.len() + 1 == self.memtable_nums as usize {
                drop(immu_tables);
                self.flush_memtable()?;
            }
            let mut immu_tables = self.immu_memtables.write();
            let new_table_id = match immu_tables.last() {
                None => self.active_memtable.read().id + 1,
                Some(table) => table.read().id + 1,
            };
            let memtable_path = PathBuf::from(&self.dir_path).join(format!("SEG.{}", new_table_id));
            let new_active_memtable = Arc::new(RwLock::new(MemTable::new(
                memtable_path,
                new_table_id,
                self.mem_table_size,
            )?));
            let mut tmp_value = new_active_memtable.clone();
            let mut foo_a = self.active_memtable.clone();
            mem::swap(&mut foo_a, &mut tmp_value);
            immu_tables.push(tmp_value);
        }

        Ok(())
    }
    pub fn put(&self, key: &[u8], value: &[u8]) -> Result<(), io::Error> {
        if key.is_empty() {
            return Ok(());
        }
        let log = LogValue {
            log_mark: LogMark::Write,
            value: Some(value.to_vec()),
        };
        self.write(key, log)
    }

    fn exist(&self, key: &[u8]) -> Result<u64, io::Error> {
        let get = self.index.get(&String::from_utf8_lossy(key).into_owned())?;
        match get {
            Some(offset) => {
                let res = bytes_to_u64(&offset);
                Ok(res)
            }
            _ => Err(io::Error::new(io::ErrorKind::NotFound, "Key not found")),
        }
    }

    pub fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, io::Error> {
        if key.is_empty() {
            return Err(io::Error::new(io::ErrorKind::NotFound, "Key is empty"));
        }
        // active memtable
        if let Some(res) = self.active_memtable.read().get(key) {
            let log = LogValue::decode(&res);
            return Ok(log.value);
        }

        // immu memtable
        for table in self.immu_memtables.read().iter() {
            if let Some(res) = table.read().get(key) {
                let log = LogValue::decode(&res);
                return Ok(log.value);
            }
        }

        // disk
        if let Ok(offset) = self.exist(key) {
            let mut db_file = self.vlog.write();
            let entry = db_file.read(offset)?;
            return Ok(Some(entry.value));
        }
        Ok(None)
    }

    pub fn del(&self, key: &[u8]) -> Result<(), io::Error> {
        if key.is_empty() {
            return Ok(());
        }
        if !self.has(key)? {
            return Ok(());
        }
        let log = LogValue {
            log_mark: LogMark::Delete,
            value: None,
        };
        self.write(key, log)
    }

    pub fn has(&self, key: &[u8]) -> Result<bool, io::Error> {
        if let Some(e) = self.active_memtable.read().get(key) {
            let res = match LogValue::decode(&e).log_mark {
                LogMark::Write => true,
                LogMark::Delete => false,
            };
            return Ok(res);
        }
        for memtable in self.immu_memtables.read().iter() {
            if let Some(e) = memtable.read().get(key) {
                let res = match LogValue::decode(&e).log_mark {
                    LogMark::Write => true,
                    LogMark::Delete => false,
                };
                return Ok(res);
            }
        }
        self.index
            .contains_key(&String::from_utf8_lossy(key).to_string())
            .map_err(|err| io::Error::new(io::ErrorKind::Other, err.to_string()))
    }

    pub fn scan_prefix(&self, prefix: &[u8]) -> Vec<Vec<u8>> {
        // 会有重复key，以active table为准
        let mut kvs: BTreeSet<Bytes> = BTreeSet::new();
        let res3: Vec<_> = self
            .index
            .iter()
            .filter(|entry| {
                let key = entry.clone().unwrap();
                let key = key.0;
                let key = bytes::Bytes::copy_from_slice(&key);
                key.starts_with(prefix)
            })
            .map(|entry| {
                let entry = entry.unwrap();
                entry.0.clone()
            })
            .collect();
        for e in res3 {
            kvs.insert(Bytes::copy_from_slice(&e));
        }
        for vec in self.immu_memtables.read().iter() {
            let res2: Vec<_> = vec
                .read()
                .inner
                .iter()
                .filter(|entry| {
                    let key = entry.key();
                    bytes::Bytes::copy_from_slice(&key).starts_with(prefix)
                })
                .map(|entry| entry.key().to_vec())
                .collect();
            for e in res2 {
                kvs.insert(Bytes::copy_from_slice(&e));
            }
        }
        let res1: Vec<_> = self
            .active_memtable
            .read()
            .inner
            .iter()
            .filter(|entry| {
                let key = entry.key();
                bytes::Bytes::copy_from_slice(&key).starts_with(prefix)
            })
            .map(|entry| entry.key().to_vec())
            .collect();
        for e in res1 {
            kvs.insert(Bytes::copy_from_slice(&e));
        }

        kvs.iter().map(|e| e.to_vec()).collect()
    }

    // flush first immutable memtable to disk
    fn flush_memtable(&self) -> io::Result<()> {
        let mut immu_tables = self.immu_memtables.write();
        let first = immu_tables[0].clone();
        let mut vlog = self.vlog.write();
        for e in first.read().inner.iter() {
            let key = e.key();
            let value = e.value();
            let log = LogValue::decode(value);
            match log.log_mark {
                LogMark::Write => {
                    let entry = Entry::new(key.to_vec(), log.value.unwrap(), 0);
                    vlog.write(&entry)?;
                }
                LogMark::Delete => {
                    let entry = Entry::new(key.to_vec(), Vec::new(), 1);
                    vlog.write(&entry)?;
                }
            }
        }
        immu_tables.remove(0);
        Ok(())
    }

    fn load_indexes_from_file(&self) -> Result<(), io::Error> {
        let mut offset = 0;
        let mut db_file = self.vlog.write();
        loop {
            match db_file.read(offset) {
                Ok(e) => {
                    let key = String::from_utf8_lossy(&e.key).into_owned();
                    if e.mark == 0 {
                        self.index.insert(key, &u64_to_bytes(offset))?;
                    } else if e.mark == 1 {
                        self.index
                            .remove(&String::from_utf8_lossy(&e.key).into_owned())?;
                    }
                    offset += e.size() as u64;
                }
                Err(err) => {
                    if err.kind() == io::ErrorKind::UnexpectedEof {
                        break;
                    } else {
                        return Err(err);
                    }
                }
            }
        }
        Ok(())
    }

    pub fn iter(&self) -> MiniBitcaskIter {
        MiniBitcaskIter::new(self)
    }
}

pub struct MiniBitcaskIter<'a> {
    vlog: Arc<RwLock<DBFile>>,
    active_table_iter: Option<crossbeam_skiplist::map::Iter<'a, Bytes, Bytes>>,
    immutable_iters: Vec<crossbeam_skiplist::map::Iter<'a, Bytes, Bytes>>,
    index_iter: Option<Box<dyn Iterator<Item = sled::Result<(IVec, IVec)>>>>,
    current_immutable_iter: usize,
}

impl<'a> MiniBitcaskIter<'a> {
    pub fn new(bitcask: &'a MiniBitcask) -> Self {
        let active_table = Box::leak(Box::new(bitcask.active_memtable.read()));
        let iter1 = active_table.inner.iter();
        let immu_tables: &'a RwLockReadGuard<'a, _> =
            Box::leak(Box::new(bitcask.immu_memtables.read()));
        let mut map2_iters = Vec::new();
        for it in immu_tables.iter() {
            let tmp: &'a RwLockReadGuard<'a, MemTable> = Box::leak(Box::new(it.read()));
            let map2_iter = tmp.inner.iter();
            map2_iters.push(map2_iter);
        }
        let xx: Box<dyn Iterator<Item = sled::Result<(IVec, IVec)>>> =
            Box::new(bitcask.index.iter());

        Self {
            vlog: bitcask.vlog.clone(),
            active_table_iter: Some(iter1),
            immutable_iters: map2_iters,
            index_iter: Some(xx),
            current_immutable_iter: 0,
        }
    }
}

impl<'a> Iterator for MiniBitcaskIter<'a> {
    type Item = (Bytes, Bytes);

    fn next(&mut self) -> Option<Self::Item> {
        // Iterate over map1
        if let Some(ref mut iter) = self.active_table_iter {
            while let Some(item) = iter.next() {
                let log = LogValue::decode(&item.value());
                match log.log_mark {
                    LogMark::Write => {
                        return Some((
                            item.key().clone(),
                            Bytes::copy_from_slice(&log.value.unwrap()),
                        ));
                    }
                    LogMark::Delete => {
                        continue;
                    }
                }
            }
        }
        self.active_table_iter = None;

        // Iterate over each BTreeMap in map2
        while self.current_immutable_iter < self.immutable_iters.len() {
            while let Some(item) = self.immutable_iters[self.current_immutable_iter].next() {
                let log = LogValue::decode(&item.value());
                match log.log_mark {
                    LogMark::Write => {
                        return Some((
                            item.key().clone(),
                            Bytes::copy_from_slice(&log.value.unwrap()),
                        ));
                    }
                    LogMark::Delete => {
                        continue;
                    }
                }
            }
            self.current_immutable_iter += 1;
        }

        // Iterate over map3
        if let Some(ref mut iter) = self.index_iter {
            if let Some(item) = iter.next() {
                let it = item.unwrap();
                let key = it.0;
                let key = Bytes::copy_from_slice(&key);
                let value = it.1;
                let value = Bytes::copy_from_slice(&value);
                let offset = bytes_to_u64(&value);
                let v = self.vlog.write().read(offset).unwrap();
                return Some((key, Bytes::from(v.value)));
            }
        }
        self.index_iter = None;

        None
    }
}

// const MEMTABLE_NUMS: i32 = 15;
// const MEMTABLE_SIZE: u32 = 64 << 20;
#[derive(Clone)]
pub struct OpenOptions {
    dir_path: String,
    memtable_nums: i32,
    mem_table_size: u32,
}

impl OpenOptions {
    pub fn new<P: AsRef<Path>>(path: P) -> Self {
        OpenOptions {
            dir_path: path.as_ref().to_string_lossy().to_string(),
            memtable_nums: 15,
            mem_table_size: 64 << 20,
        }
    }

    pub fn memtable_nums(&self, nums: i32) -> Self {
        let mut op = self.clone();
        op.memtable_nums = nums;
        op
    }
    pub fn memtable_size(&self, sz: u32) -> Self {
        let mut op = self.clone();
        op.mem_table_size = sz;
        op
    }

    pub fn open(&self) -> io::Result<MiniBitcask> {
        if let Err(err) = fs::create_dir_all(&self.dir_path) {
            if err.kind() != io::ErrorKind::AlreadyExists {
                return Err(err);
            }
        }

        let dir_abs_path = fs::canonicalize(&self.dir_path)?;
        let db_file_path = dir_abs_path.to_string_lossy().to_string();
        let db_file = Arc::new(RwLock::new(DBFile::new(&db_file_path)?));
        let idx_path = PathBuf::from(&self.dir_path).join("idx");
        let indexes = sled::open(idx_path)?;
        let memtable_path = PathBuf::from(&self.dir_path).join("SEG.0");

        let mini_bitcask = MiniBitcask {
            active_memtable: Arc::new(RwLock::new(MemTable::new(
                memtable_path,
                0,
                self.mem_table_size,
            )?)),
            immu_memtables: Arc::new(RwLock::new(Vec::new())),

            index: indexes,
            vlog: db_file,
            dir_path: dir_abs_path.to_string_lossy().into_owned(),
            memtable_nums: self.memtable_nums,
            mem_table_size: self.mem_table_size,
        };
        // TODO recover memtables from wal

        mini_bitcask.load_indexes_from_file()?;
        Ok(mini_bitcask)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use temp_dir::TempDir;

    #[test]
    fn test1() {
        let d = TempDir::new().unwrap();
        let db_path = PathBuf::from(d.path()).join("test1");
        let db = MiniBitcask::open(&db_path).unwrap();
        db.put(b"a", b"a").unwrap();
        db.put(b"b", b"b").unwrap();
        let a = db.get(b"a").unwrap().unwrap();
        assert_eq!(&a, b"a");
        let b = db.get(b"b").unwrap().unwrap();
        assert_eq!(&b, b"b");
        db.del(b"a").unwrap();
        let a = db.get(b"a").unwrap();
        assert!(a.is_none());
    }

    #[test]
    fn test2() {
        let d = TempDir::new().unwrap();
        let db_path = PathBuf::from(d.path());
        let db = MiniBitcask::open(&db_path).unwrap();
        db.put(b"a", b"a").unwrap();
        db.put(b"a1", b"a").unwrap();
        db.put(b"a2", b"a").unwrap();
        db.put(b"b", b"b").unwrap();
        let res = db.scan_prefix(b"a");
        assert_eq!(res.len(), 3);
    }
    #[test]
    fn test3() {
        let d = TempDir::new().unwrap();
        let db_path = PathBuf::from(d.path());
        let db = MiniBitcask::open(&db_path).unwrap();
        db.put(b"a", b"a").unwrap();
        db.put(b"a1", b"a").unwrap();
        db.put(b"a2", b"a").unwrap();
        db.put(b"b", b"b").unwrap();
        let vec: Vec<_> = db.iter().collect();
        assert_eq!(vec.len(), 4);
        for it in vec {
            println!(
                "{} {}",
                String::from_utf8_lossy(&it.0),
                String::from_utf8_lossy(&it.1)
            );
        }
    }
}
